package com.ghca.storm;

import org.apache.storm.shade.org.apache.commons.io.FileUtils;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;

/**
 * Created by gh on 2016/10/24.
 */
public class DataSourceSpout extends BaseRichSpout {

    private static final long serialVersionUID = 1L;

    private Map conf;
    private TopologyContext context;
    private SpoutOutputCollector collector;

    /**
     * 初始化方法： 本实例运行的时候执行一次且仅一次
     */
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.conf = conf;
        this.context = context;
        this.collector = collector;
    }

    /**
     * 死循环调用： tuple 为storm传输数据基本单位
     */
    public void nextTuple() {

        // 读取文件列表
        Collection<File> listFiles = FileUtils.listFiles(new File("d:/test"), new String[] { "txt" }, true);
        // 循环每个文件
        for (File file : listFiles) {
            // 行格式发送
            try {
                List<String> lines = FileUtils.readLines(file);
                for (String line : lines) {
                    this.collector.emit(new Values(line));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            // 文件已经处理完成
            try {
                File srcFile = file.getAbsoluteFile();
                File destFile = new File(srcFile + ".done." + System.currentTimeMillis());
                FileUtils.moveFile(srcFile, destFile);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("line"));
    }
}
